package edu.upenn.seas.peerreview.transport

import scala.collection.mutable
import scala.util.logging.Logged

import edu.upenn.seas.peerreview.{Id, Sink, Source}

/**
 * Implements a local (single-JVM) transport layer.
 */
class LocalTransport extends Transport with Logged {
  
  private val validIds = mutable.Set[Id](Id.coordinatorId)
  private val sources = mutable.Map[Id, Source.T]()
  private val mesgQueue = new mutable.SynchronizedQueue[(Id, Id, Message)]()
  private var processing = false
  private var paused = false
  
  override def register(name: String, factory: Source.Factory): Id = {
    assert(name != Transport.coordinatorName
	   || !sources.contains(Id.coordinatorId),
           "coordinator already registered")
    val id = LocalTransport.newLocalId(name)
    validIds += id
    sources(id) = factory(id, enqueueMesg(id, _, _))
    assert(sources(id) != null, "sources must not be null")
    log("Registered " + id)
    processQueue()
    id
  }
  
  private def processQueue() {
    val proceed = synchronized {
      val proceed = !processing
      processing = true
      proceed
    }
    if (proceed) {
      while (mesgQueue.size > 0 && sources.contains(mesgQueue.front._2)) {
	// note this doesn't check the validity of the "from" id
	val (from, to, mesg) = mesgQueue.dequeue()
	log("%s receives from %s: %s".format(to, from, mesg))
	sources(to)(from, mesg)
      }
      processing = false
    }
  }
  
  private[transport] def enqueueMesg(from: Id, to: Id, mesg: Message) {
    assert(validIds.contains(to), "unrecognized id=[" + to + "]")
    log("%s sends to %s: %s".format(from, to, mesg))
    mesgQueue += (from, to, mesg)
    processQueue()
  }
  
  private def assert(assertion: Boolean, message: String) =
    Predef.assert(assertion, "%s [MesgQueue: %s]".format(message, mesgQueue))
  
  /**
   * For testing only.
   */
  private[transport] def setPaused(paused: Boolean) {
    this.paused = paused
    processQueue()
  }
}

object LocalTransport {

  private var nextIdNumber = 0
  
  private def newLocalId(name: String) = synchronized {
    // synchronized for use by multiple LocalTransport instances
    if (name == Transport.coordinatorName) {
      Id.coordinatorId
    } else {
      nextIdNumber += 1
      LocalTransport.LocalId(name, nextIdNumber)
    }
  }
  
  final case class LocalId(name: String, id: Int) extends Id {
    override def toString = name + id
  }
}
